
package com.flinkBigWork.alltask;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.*;
import java.util.Properties;

public class kafkapro_true {
    public static void main(String[] args) {
        // 设置Kafka的生产者配置
        Properties props = new Properties();
        // 1. 指定Kafaka集群的ip地址和端口号
        props.put("bootstrap.servers", "192.168.88.101:19092,192.168.88.101:29092,192.168.88.101:39092");
//        // 2. 等待所有副本节点的应答
//        props.put("acks", "all");
//        // 3. 消息发送最大尝试次数
//        props.put("retries", 0);
//        // 4. 指定一批消息处理次数
//        props.put("batch.size", 16384);
//        // 5. 指定请求延时
//        props.put("linger.ms", 1);
//        // 6. 指定缓存区内存大小
//        props.put("buffer.memory", 33554432);
        // 7. 设置key序列化
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 8. 设置value序列化
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 创建 KafkaProducer 实例
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        BufferedReader reader = null;

        try {
            // 替换成你实际的文件路径
            String path1="D:\\新建文件夹 (3)\\2024大三上学期文件夹\\大数据课设\\optimized_canteen_data.csv";
            // 1. 打开文件进行读取，使用GBK编码
            File file = new File(path1); // 替换成你实际的文件路径
            reader = new BufferedReader(new InputStreamReader(new FileInputStream(file)));

            String line;

            String topic = "foo";  // 替换为你使用的 Kafka 主题


            // 2. 按行读取文件并发送每行作为消息
            while ((line = reader.readLine()) != null) {
                // 生成一个消息记录
                ProducerRecord<String, String> record = new ProducerRecord<>(topic, null, line);

                // 打印消息内容（可选）
                System.out.println("Sending message: " + line);

                // 3. 发送消息
                producer.send(record, (metadata, exception) -> {
                    if (exception != null) {
                        System.err.println("Error sending message: " + exception.getMessage());
                    } else {
                        System.out.println("Message sent to topic " + metadata.topic() + " partition " + metadata.partition() + " at offset " + metadata.offset());
                    }
                });
            }

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            // 关闭文件读取流
            try {
                if (reader != null) {
                    reader.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }

            // 关闭 Kafka producer
            producer.close();
        }
    }
}